package com.hzgj.bcl.soa.cicada.pool;

import com.hzgj.bcl.soa.config.ChannelConfig;
import com.hzgj.bcl.soa.rpc.IpAddress;
import com.hzgj.bcl.soa.util.NamedThreadFactory;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.pool.ChannelHealthChecker;
import io.netty.channel.pool.ChannelPoolHandler;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.Promise;

public class DefaultChannelPool implements StatisticChannelPool {

    Bootstrap bootstrap = new Bootstrap();
    private Logger logger = LoggerFactory.getLogger(getClass());
    private FixedChannelPool pool;
    private IpAddress address;
    private int maxSize;

    public DefaultChannelPool(ChannelPoolConfig channelPoolConfig, IpAddress ipAddress, ChannelConfig channelConfig, ChannelPoolHandler channelPoolHandler) {
        this.address = ipAddress;
        bootstrap = createBootstrap(channelConfig);
        maxSize = channelPoolConfig.getMaxConnections();
        pool = new FixedChannelPool(bootstrap,
                channelPoolHandler,
                ChannelHealthChecker.ACTIVE,
                FixedChannelPool.AcquireTimeoutAction.FAIL,
                channelPoolConfig.getAcquireTimeoutMillis(),
                channelPoolConfig.getMaxConnections(),
                channelPoolConfig.getMaxPendingAcquires());
    }

    private Bootstrap createBootstrap(ChannelConfig channelConfig) {
        Bootstrap bootstrap = new Bootstrap();

        bootstrap.group(new NioEventLoopGroup(channelConfig.getEventThreads(), new NamedThreadFactory("client-worker-")))
                .channel(NioSocketChannel.class)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .option(ChannelOption.TCP_NODELAY, Boolean.parseBoolean(System.getProperty("com.hzgj.soa.tcp.nodelay", "true")))
                .option(ChannelOption.SO_REUSEADDR, Boolean.parseBoolean(System.getProperty("com.hzgj.soa.tcp.reuseaddress", "true")))
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, channelConfig.getConnectTimeoutMills());
//                .option(ChannelOption.SO_KEEPALIVE, options.isKeepAlive());
//                .option(ChannelOption.TCP_NODELAY, options.isTcpNoDelay());
//                .option(ChannelOption.MESSAGE_SIZE_ESTIMATOR, new DefaultMessageSizeEstimator(options.getReceiveBufferSize()));
        bootstrap.remoteAddress(address.getIpAddress(), address.getPort());
        return bootstrap;
    }

    @Override
    public Future<Channel> acquire() {
        return pool.acquire();
    }

    @Override
    public Future<Channel> acquire(Promise<Channel> promise) {
        return pool.acquire(promise);
    }

    @Override
    public Future<Void> release(Channel channel) {
        return pool.release(channel);
    }

    @Override
    public Future<Void> release(Channel channel, Promise<Void> promise) {
        return pool.release(channel, promise);
    }

    @Override
    public void close() {
        pool.close();
        bootstrap.config().group().shutdownGracefully();
    }

    @Override
    public int getMaxSize() {
        return maxSize;
    }

    @Override
    public int getWorkingSize() {
        return pool.getAcquiredChannelCount();
    }

    public static abstract class SimpleChannelPoolHandler implements ChannelPoolHandler {
        private static Logger logger = LoggerFactory.getLogger(SimpleChannelPoolHandler.class);

        @Override
        public void channelReleased(Channel channel) throws Exception {
            if (logger.isTraceEnabled()) {
                logger.trace("channelReleased");
            }
        }

        @Override
        public void channelAcquired(Channel channel) throws Exception {
            if (logger.isTraceEnabled()) {
                logger.trace("channelAcquired");
            }
        }
    }
}
